package Flink.wc;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketStreamWordCount {
    // Todo Flink 无界流处理

    public static void main(String[] args) throws Exception {
        // 1. 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 读取文本流：hadoop102表示发送端主机名、7777表示端口号
        DataStreamSource<String> lineDS = env.socketTextStream("master", 7777);

        // 3. 转换、分组、求和，得到统计结果
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineDS.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                    /*
                    Flink 还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，
从而获得对应的序列化器和反序列化器。但是，由于 Java中泛型擦除的存在，在某些特殊情
况下（比如 Lambda 表达式中），自动提取的信息是不够精细的——只告诉 Flink 当前的元素
由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类
型信息，才能使应用程序正常工作或提高其性能。
因为对于flatMap里传入的Lambda表达式，系统只能推断出返回的是Tuple2类型，而无
法得到Tuple2<String, Long>。只有显式地告诉系统当前的返回类型，才能正确地解析出完整
数据。
                     */
                }).returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(data -> data.f0)
                .sum(1);

        // 4. 打印
        sum.print();

        // 5. 执行
        env.execute();
        /*
          我们会发现程序启动之后没有任何输出、也不会退出。这是正常的，因为Flink的流处理
          是事件驱动的(区别于Spark Streaming)，当前程序会一直处于监听状态，只有接收到数据才会执行任务、输出统计结果。
         */
    }
}
